package top.wangjiu.webflux;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.net.URI;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 这个项目是为了证明spring的webflux不支持回压功能
 */
@SpringBootApplication
@RestController
public class WebfluxApplication {

    private static final Log log = LogFactory.getLog(WebfluxApplication.class);

    private CountDownLatch countDownLatch = new CountDownLatch(1);
    
    public static void main(String[] args) throws InterruptedException {
        ConfigurableApplicationContext context = SpringApplication.run(WebfluxApplication.class, args);
        WebfluxApplication app = context.getBean(WebfluxApplication.class);
        app.getCount();
        //这个测试内容是，生产者总共生产10个数据，消费者每次请求1个数据，一共请求10次，
        //我期望的效果是，消费者请求一次，生产者生产一个数据，并立刻返回，消费者消费完之后请求下一个数据，直到生产者生产结束。

        //但通过输出可以看到，生产者是生产完全部数据后，才将全部数据一次发给消费者，不符合我期望的效果。
        //并且从生产者日志可以看到，生产者收到的request并不是 1，而是 9223372036854775807，说明消费者调用 Subscriber api完全没有用。
        app.countDownLatch.await();
        log.info("测试结束，webflux不支持回压！");
        context.close();
    }


    @GetMapping("test1")
    public Mono<Void> getCount(){
        WebClient client = WebClient.builder().build();
        Flux<Integer> res = client.get().uri("http://localhost:8088/test").retrieve().bodyToFlux(Integer.class);
        log.info("request:"+Thread.currentThread().getId());
        // 线程Id不同，证明是异步，
        // 但是服务提供者并没有按照Subscription设置的3进行响应，而且是服务端将响应全部处理完，客户端才收到全部内容，说明不支持全链路的回压可控
        res.subscribe(new Subscriber<Integer>() {
            Subscription s = null;
            @Override
            public void onSubscribe(Subscription s) {
                this.s = s;
                s.request(1);
                log.info("onSubscribe:"+Thread.currentThread().getId());
            }

            @Override
            public void onNext(Integer integer) {
                log.info("onNext:"+Thread.currentThread().getId()+",res:"+integer);
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                s.request(1);
            }

            @Override
            public void onError(Throwable t) {

            }

            @Override
            public void onComplete() {
                countDownLatch.countDown();
            }
        });
        return Mono.empty();
    }

    @GetMapping("test")
    public Flux<Integer> productTest(){
        log.info("收到请求");
        AtomicInteger num = new AtomicInteger(0);
        return Flux.create(fluxSink -> {
            if (fluxSink.isCancelled()) {
                fluxSink.complete();
                return;
            }
            fluxSink.onRequest(r -> {
                log.info("product receive request count:"+r);
                int i = 0;
                while (i++ < r && num.getAndIncrement() < 10) {
                    fluxSink.next(num.get());
                    log.info("product:" + num.get());
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                fluxSink.complete();
            });
        });
    }

}
